import csv
import os
import time
from subprocess import Popen
from typing import Dict
from uuid import uuid1

import pandas as pd
import yaml
from notification_service.client import NotificationClient
from notification_service.base_notification import EventWatcher, BaseEvent
from kafka import KafkaProducer, KafkaAdminClient, KafkaConsumer
from kafka.admin import NewTopic
from typing import List
import sys, getopt


class KafkaWatcher(EventWatcher):
    def __init__(self, bootstrap_servers, input_filename, input_topic):
        super().__init__()
        self.bootstrap_servers = bootstrap_servers
        self.input_filename = input_filename
        self.input_topic = input_topic

    def process(self, events: List[BaseEvent]):
        print("watcher event triggered")
        time.sleep(30)
        
        producer = KafkaProducer(
            bootstrap_servers = self.bootstrap_servers, 
            value_serializer=lambda v: v.encode())
        
        f = open(self.input_filename)
        while True:
            line = f.readline()
            if not line:
                break
            line = line.strip()
            future = producer.send(self.input_topic,value=line)
            time.sleep(0.002)
        time.sleep(20)


def init_kafka(bootstrap_servers, input_topic, output_topic):
    admin_client = KafkaAdminClient(bootstrap_servers=bootstrap_servers)
    topics = admin_client.list_topics()
    if input_topic not in topics:
        print("create input topic: "+input_topic)
        admin_client.create_topics(
            new_topics=[NewTopic(name=input_topic, num_partitions=1, replication_factor=1)])
    if output_topic not in topics:
        print("create output topic: "+output_topic)
        admin_client.create_topics(
            new_topics=[NewTopic(name=output_topic, num_partitions=1, replication_factor=1)])


if __name__ == '__main__':
    opts, args = getopt.getopt(sys.argv[1:],"",["input_filename=","input_topic=","output_topic=","server="])  
    mydict = dict(opts)
    input_filename = mydict.get('--input_filename', '')
    input_topic = mydict.get('--input_topic', '')
    output_topic = mydict.get('--output_topic', '')
    bootstrap_servers = mydict.get('--server', '')
    bootstrap_servers = bootstrap_servers.split(',')
    print((input_filename, input_topic, output_topic, bootstrap_servers, ))
    init_kafka(bootstrap_servers, input_topic, output_topic)

    notification_client = NotificationClient('localhost:50051', default_namespace="default")
    notification_client.start_listen_event(key='KafkaWatcher', event_type = 'UNDEFINED', namespace="default", 
        watcher=KafkaWatcher(bootstrap_servers, input_filename, input_topic),
        start_time=0)


